package edu.hhu.wifispark.task.sparktask;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;
import edu.hhu.wifispark.entry.BasePhone;
import edu.hhu.wifispark.entry.Customerflow;
import edu.hhu.wifispark.entry.Customerlive;
import edu.hhu.wifispark.util.SparkProUtil;

public class CFlowTask extends Thread implements Serializable{
	private static final long serialVersionUID = 1L;
	public List<BasePhone> bps;
	public long atime = 0l;
	String taskName= null;
	public CFlowTask(){
		this.taskName = "instoreRate";
	}
	public CFlowTask(long time){
		this.taskName = "instoreRate"+time;
		this.atime = time;
	}
	
	@SuppressWarnings("resource")
	public void run(){
		SparkConf sparkConf = new SparkConf().setAppName(taskName).setMaster(SparkProUtil.MASTER);
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		JavaRDD<BasePhone> phoneRDD = jsc.parallelize(bps);

		
		JavaPairRDD<String, Integer> phonep = phoneRDD.mapToPair(new PairFunction<BasePhone, String, Integer>() {
			private static final long serialVersionUID = 1L;

			public Tuple2<String, Integer> call(BasePhone arg0)
					throws Exception {
				if(arg0.getInt("inner") == 1){
					return new Tuple2<String, Integer>(arg0.getStr("mac"), 2);
				}else if(arg0.getInt("outer") == 1){
					return new Tuple2<String, Integer>(arg0.getStr("mac"), 1);
				}else{
					return new Tuple2<String, Integer>(arg0.getStr("mac"), 0);
				}
			}
		});
		
		JavaPairRDD<String, Integer> res0 = phonep.reduceByKey(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 1L;

			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0>arg1?arg0:arg1;
			}
		});

//		JavaPairRDD<String, Integer> res1 = res0.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {
//			private static final long serialVersionUID = 1L;
//
//			public Tuple2<String, Integer> call(Tuple2<String, Integer> t)
//					throws Exception {
//				
//				return new Tuple2<String, Integer>(t._1.split(",")[1], t._2);
//			}
//		});
		
		JavaRDD<Integer> res1 = res0.map(new Function<Tuple2<String,Integer>, Integer>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Integer call(Tuple2<String, Integer> v1) throws Exception {
				return v1._2;
			}
		});

		Map<Integer, Long> res2 = res1.countByValue();
		jsc.stop();
		System.out.println(res2);
		Customerflow flow = new Customerflow();
		flow.set("atime", atime);
		flow.set("innernum", res2.get(2).intValue());
		flow.set("outernum", res2.get(1).intValue());
		int oldCount = Customerlive.dao.selectOldCustomer(atime);
		flow.set("oldcustomer", oldCount).save();
	}
	public void setBps(List<BasePhone> ibps){
		this.bps = ibps;
	}
}
